//package org.bulu.project.web.config;
//
//import com.alibaba.fastjson.JSONArray;
//import com.alibaba.fastjson.JSONObject;
//import io.netty.util.internal.StringUtil;
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
//import org.apache.rocketmq.client.producer.DefaultMQProducer;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.bulu.project.common.constant.UserMomentsConstant;
//import org.bulu.project.system.model.entity.Dynamic;
//import org.bulu.project.system.model.query.UserFollowQuery;
//import org.bulu.project.system.model.vo.UserFollowVO;
//import org.bulu.project.system.service.FollowService;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.context.annotation.Bean;
//import org.springframework.context.annotation.Configuration;
//import org.springframework.data.redis.core.RedisTemplate;
//
//import javax.annotation.Resource;
//import java.util.ArrayList;
//import java.util.List;
//
////rocketmq配置文件
//@Configuration
//public class RocketMQconfig {
//
//    @Value("${apache.rocketmq.name.server}")
//    private String nameServerAddr;
//
//    @Resource
//    private RedisTemplate<String, String> redisTemplate;
//
//    @Autowired
//    private FollowService followService;
//
//    @Bean("momentsProducer")
//    public DefaultMQProducer momentsProducer() throws Exception {
//        DefaultMQProducer producer = new DefaultMQProducer(UserMomentsConstant.GROUP_MOMENTS);
//        producer.setNamesrvAddr(nameServerAddr);
//        producer.start();
//        return producer;
//    }
//
//
//    @Bean(" momentsConsumer")
//    public DefaultMQPushConsumer momentsConsumer() throws Exception {
//        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(UserMomentsConstant.GROUP_MOMENTS);
//        consumer.setNamesrvAddr(nameServerAddr);
//        consumer.subscribe(UserMomentsConstant.TOPIC_MOMENTS, "*");//订阅
//        //监听器监听到消息，然后push操作
//        consumer.registerMessageListener(new MessageListenerConcurrently() {
//            @Override//获取到消息，以及上下文
//            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
//                MessageExt msg = msgs.get(0);//如果消息为空，则返回成功标识
//                if (msg == null) {
//                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//                }
//                String bodyStr = new String(msg.getBody());//获取消息，但是是字符串形式的
//                Dynamic dynamic = JSONObject.toJavaObject(JSONObject.parseObject(bodyStr), Dynamic.class);
//                //Integer userId=dynamic.getUid();//得到用户id
//                UserFollowQuery queryParams = new UserFollowQuery();
//                List<UserFollowVO> fanlist = (List<UserFollowVO>) followService.getFansList(queryParams);//通过用户id知道用户的粉丝列表
//                for (UserFollowVO fan : fanlist) {//遍历粉丝列表
//                    String key = "subscribed" + fan.getUserId();//设置redis的键值为粉丝id+消息订阅字样
//                    String subscribedListStr = redisTemplate.opsForValue().get(key);//获取指定key对应的值,字符串形式
//                    List<Dynamic> subscribedList;
//                    if (StringUtil.isNullOrEmpty(subscribedListStr)) {//如果值为空
//                        subscribedList = new ArrayList<>();//新建一个数组List集合
//                    } else {
//                        subscribedList = JSONArray.parseArray(subscribedListStr, Dynamic.class);//不为空，则将得到的值转换成usermoment的数组集合
//                    }
//                    subscribedList.add(dynamic);//在将dynamic信息加入数组集合中
//                    redisTemplate.opsForValue().set(key, JSONObject.toJSONString(subscribedList));//在redis中设置key,values形式键值对，values又需要转换成String类型
//                }
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//返回成功字样
//            }
//        });
//        consumer.start();
//        return consumer;
//    }
//}